package pulsar

import (
	"knowledge_admin/pkg/adapters/queue/pubsub"
	"knowledge_admin/pkg/framework"
	"knowledge_admin/pkg/utils"
)

// GetSubscriber get subscriber from cache.
func (c *PulsarClient) GetSubscriber(name string) pubsub.Subscriber {
	return c.subscriberCache[name]
}

// GetPublisher get publisher from cache.
func (c *PulsarClient) GetPublisher(name string) pubsub.Publisher {
	return c.publisherCache[name]
}

type PulsarClient struct {
	subscriberCache map[string]pubsub.Subscriber
	publisherCache  map[string]pubsub.Publisher
}

func NewPulsarClient(cfg framework.Configuration) (*PulsarClient, error) {
	conf := utils.LoadConfig(cfg, Config{}).(Config)
	pubConfigs := conf.Publishers
	subConfigs := conf.Subscribers
	subscriberCache := map[string]pubsub.Subscriber{}
	publisherCache := map[string]pubsub.Publisher{}
	for key, pubOption := range pubConfigs {
		publisher, err := NewPublisher(pubOption)
		if err != nil {
			return nil, err
		}
		publisherCache[key] = publisher
	}
	for key, subOption := range subConfigs {
		subscriber, err := NewSubscriber(subOption)
		if err != nil {
			return nil, err
		}
		subscriberCache[key] = subscriber
	}
	return &PulsarClient{
		subscriberCache: subscriberCache,
		publisherCache:  publisherCache,
	}, nil
}
